package com.zp.self.module.level_4_算法练习.算法.多线程;

import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author ：by Zeng Peng
 * @date ：Created 2022/9/1 20:00
 * @description ：
 */
public class Condition案例_生产与消费 {
    final ReentrantLock lock = new ReentrantLock();
    //生产者锁定，库存满了，进行等待消费，一般不会进入等待搁置
    final AbstractQueuedSynchronizer.ConditionObject fullAwait = (AbstractQueuedSynchronizer.ConditionObject) lock.newCondition();
    //消费者锁定，库存没有，锁定进行等待
    final AbstractQueuedSynchronizer.ConditionObject emptyAwait = (AbstractQueuedSynchronizer.ConditionObject) lock.newCondition();

    final Product[] items = new Product[10];
    int putptr, takeptr, count;

    public void put(Product x) throws InterruptedException {
        lock.lock();
        try {
            // 当数组满了
            while (count == items.length) {
                System.out.println("库存满了：停工等待消费者消费.count = "+ count);
                fullAwait.await(); // 释放锁，等待
            }
            // 放入数据
            items[putptr] = x;
            // 如果到最后一个位置了,下标从头开始,防止下标越界， putptr从头开始
            if (++putptr == items.length)
                putptr = 0;
            // 对 count ++ 加加
            ++count;
            // 通知 take 线程,可以取数据了,不必继续阻塞
            emptyAwait.signal();
        } finally {
            lock.unlock();
        }
    }

    public Object take() throws InterruptedException {
        lock.lock();
        try {
            // 如果数组没有数据,则等待
            while (count == 0) {
                System.out.println("消费者:没有库存，我正在等待.....count= "+count);
                emptyAwait.await();
                System.out.println("消费者:有库存了，我被唤醒了.count= "+count);

            }
            // 取数据
            Object x = items[takeptr];
            // 如果到数组尽头了,就从头开始
            if (++takeptr == items.length) {
                // 从头开始
                takeptr = 0;
            }
            // 将数量减1
            --count;
            // 通知阻塞的 put 线程可以装填数据了
            fullAwait.signal();
            return x;
        } finally {
            lock.unlock();
        }
    }


    final static ExecutorService fixedExecutor = Executors.newFixedThreadPool(3);
    public static void main(String[] args) throws InterruptedException {

        CountDownLatch countDownLatch =new CountDownLatch(2);
        Condition案例_生产与消费 案例_生产与消费 = new Condition案例_生产与消费();

        //消费者代码：每1秒消费一个产品，消费10个结束线程
        fixedExecutor.execute(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                   System.out.println("消费者:消费了一条数据"+案例_生产与消费.take());
                    System.out.println("---------------------- 分割线 ----------------------");
                   Thread.sleep(1000);
               }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            countDownLatch.countDown();
        });


        //生产者代码：每2秒生产一个产品
        fixedExecutor.execute(() -> {
            try {
                while (true){
                    案例_生产与消费.put(new Product());
                    System.out.println("生产者:生产了一个产品");
                    Thread.sleep(2000);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            countDownLatch.countDown();
        });
        System.out.println("任务开始了");

        countDownLatch.await();

        System.out.println("任务结束了");
        fixedExecutor.shutdown();
    }
}

class Product {
    public static Random random = new Random();
    public Product (){
        this.serialNumber = "SN:"+random.nextInt(10000);
    }
    String serialNumber;
    public String getSerialNumber() {
        return serialNumber;
    }
    @Override
    public String toString() {
        return "Product{" +
                "serialNumber='" + serialNumber + '\'' +
                '}';
    }
}